package com.vhall.component.service.redpacket.tasks;

import com.alibaba.fastjson.JSON;
import com.vhall.component.entity.income.IncomeCreateDTO;
import com.vhall.component.entity.order.OrderDetailEntity;
import com.vhall.component.entity.rebroadcast.RedPacketRecordEntity;
import com.vhall.component.plugin.core.service.PaasService;
import com.vhall.component.service.order.IncomeService;
import com.vhall.component.service.order.OrderService;
import com.vhall.component.service.redpacket.service.RedPacketRecordService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

/**
 * Redis 队列消息订阅
 *
 * @author guoxin.wang
 * @date 2021/6/11
 */
@Slf4j
@Component
public class RPStreamListener implements StreamListener<String, ObjectRecord<String, String>> {

    @Value("${vhall.red-packet.stream-name:rp-group}")
    private String groupName;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private RedPacketRecordService recordService;

    @Autowired
    private PaasService paasService;

    @Autowired
    private OrderService orderService;

    @Autowired
    private IncomeService incomeService;

    @Autowired
    private RPStreamListener self;

    @Override
    public void onMessage(ObjectRecord<String, String> message) {
        // 消息ID
        RecordId messageId = message.getId();
        // 消息的key和value
        log.info("[红包组件]接收到Stream消息,messageId={}, stream={}, body={}",
                messageId, message.getStream(), message.getValue());
        // 开始处理消息的逻辑
        String value = message.getValue();

        if (StringUtils.isNotEmpty(value)) {
            self.saveRecord(value);
        }

        // 通过RedisTemplate手动确认消息
        redisTemplate.opsForStream().acknowledge(groupName, message);
    }

    @Transactional(rollbackFor = Exception.class)
    public void saveRecord(String value) {
        // 消费 添加红包领取明细,
        RedPacketRecordEntity records = JSON.parseObject(value, RedPacketRecordEntity.class);
        recordService.add(records);
        log.info("[红包组件]红包明细创建成功,record_id={}", records.getId());
        // 收入订单
        OrderDetailEntity order = new OrderDetailEntity();
        order.setRoomId(records.getRoomId());
        order.setAmount(records.getAmount());
        order.setSource(0);
        order.setTradeNo(records.getTradeNo());
        order.setAppId(paasService.getAppId());
        order.setAccountId(records.getUserId());
        orderService.create(order);
        log.info("[红包组件]订单创建成功,order_id={}", order.getId());

        IncomeCreateDTO income = new IncomeCreateDTO();
        income.setAmount(records.getAmount());
        income.setAppId(paasService.getAppId());
        income.setAccountId(records.getUserId());

        incomeService.create(income);
        log.info("[红包组件]收入表创建成功,income_id={}", income.getId());
    }

}
